/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.tracer.core;

import com.lvyh.lightframe.tracer.core.common.TraceContants;
import com.lvyh.lightframe.tracer.core.listener.SpanReportListener;
import com.lvyh.lightframe.tracer.core.listener.SpanReportListenerHolder;
import com.lvyh.lightframe.tracer.core.samplers.Sampler;
import io.opentracing.Scope;
import io.opentracing.ScopeManager;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.Tracer;
import io.opentracing.propagation.BinaryExtract;
import io.opentracing.propagation.BinaryInject;
import io.opentracing.propagation.Format;
import io.opentracing.propagation.TextMapExtract;
import io.opentracing.propagation.TextMapInject;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Simpletracer implements the tracer interface and has the ability to build span, inject and extract data.
 * <p>
 * Start a new Span: create a new span.
 * Create a new span with the specified operationname.
 * Operationname indicates the specific work done by the span (for example, RPC method name, function name, or the name of a phase in a large computing task).
 * <p>
 * Inject a SpanContext: inject SpanContext into a carrier of a given type for cross process transmission.
 * Extract a SpanContext: extract the SpanContext instance object from the carrier.
 */
public class SimpleTracer implements Tracer {
    /**
     * normal root spanId's default value
     */
    public static final String ROOT_SPAN_ID = "0";
    /**
     * Sampler instance
     */
    private Sampler sampler;

    /**
     * Cache some information related to the tracer globally
     */
    private Map<String, Object> tracerTags = new ConcurrentHashMap<>();

    private final ScopeManager scopeManager;

    public SimpleTracer() {
        ScopeManager scopeManager = null;
        try {
            Iterator<ScopeManager> it = ServiceLoader.load(ScopeManager.class).iterator();
            if (it.hasNext()) {
                scopeManager = it.next();
            }
        } catch (Exception ignored) {
        }
        if (scopeManager == null) {
            this.scopeManager = new InactiveScopeManager();
        } else {
            this.scopeManager = scopeManager;
        }
    }

    public SimpleTracer(Sampler sampler, ScopeManager scopeManager, Map<String, Object> tracerTags) {
        this.scopeManager = scopeManager;
        this.tracerTags = tracerTags;
        this.sampler = sampler;
    }

    public void reportSpan(SimpleSpan span) {
        if (span == null) {
            return;
        }
        //invoke listener
        this.invokeReportListeners(span);
    }

    protected void invokeReportListeners(SimpleSpan simpleSpan) {
        List<SpanReportListener> listeners = SpanReportListenerHolder.getSpanReportListenersHolder();
        if (listeners != null && listeners.size() > 0) {
            for (SpanReportListener listener : listeners) {
                listener.onSpanReport(simpleSpan);
            }
        }
    }

    @Override
    public SpanBuilder buildSpan(String operationName) {
        return new SimpleSpanBuilder(scopeManager, operationName);
    }

    @Override
    public ScopeManager scopeManager() {
        return scopeManager;
    }

    public Sampler getSampler() {
        return sampler;
    }

    public void setSampler(Sampler sampler) {
        this.sampler = sampler;
    }

    public Map<String, Object> getTracerTags() {
        return tracerTags;
    }

    public void setTracerTags(Map<String, Object> tracerTags) {
        this.tracerTags = tracerTags;
    }

    @Override
    public Span activeSpan() {
        return scopeManager.activeSpan();
    }

    @Override
    public Scope activateSpan(Span span) {
        return scopeManager.activate(span);
    }

    @Override
    public <C> void inject(SpanContext spanContext, Format<C> format, C carrier) {
        if (format.equals(Format.Builtin.TEXT_MAP) || format.equals(Format.Builtin.TEXT_MAP_INJECT)
                || format.equals(Format.Builtin.HTTP_HEADERS)) {
            if (!(carrier instanceof TextMapInject)) {
                throw new IllegalArgumentException("Expected text map carrier: " + carrier);
            }
            for (Map.Entry<String, String> entry : spanContext.baggageItems()) {
                ((TextMapInject) carrier).put(entry.getKey(), entry.getValue());
            }

            SimpleSpanContext simpleSpanContext = (SimpleSpanContext) spanContext;
            ((TextMapInject) carrier).put(TraceContants.TRACE_ID_ALIAS, simpleSpanContext.getTraceId());
            ((TextMapInject) carrier).put(TraceContants.SPAN_ID_ALIAS, simpleSpanContext.getSpanId());
            ((TextMapInject) carrier).put(TraceContants.PARENT_SPAN_ID_ALIAS, simpleSpanContext.getParentId());

        } else if (format.equals(Format.Builtin.BINARY) || format.equals(Format.Builtin.BINARY_INJECT)) {
            if (!(carrier instanceof BinaryInject)) {
                throw new IllegalArgumentException("Expected a byte buffer carrier: " + carrier);
            }
            int requiredSize = 1; // we end with a NO_ENTRY marker
            ArrayList<byte[]> binary = new ArrayList<byte[]>();
            for (Map.Entry<String, String> entry : spanContext.baggageItems()) {
                requiredSize += 1 + 4 + 4; // ENTRY marker + size of key and size of value
                byte[] key = entry.getKey().getBytes(ByteBufferSpanContext.CHARSET);
                byte[] value = entry.getValue().getBytes(ByteBufferSpanContext.CHARSET);
                requiredSize += key.length + value.length;
                binary.add(key);
                binary.add(value);
            }
            ByteBuffer injectionBuffer = ((BinaryInject) carrier).injectionBuffer(requiredSize);
            Iterator<byte[]> iterator = binary.iterator();
            while (iterator.hasNext()) {
                byte[] key = iterator.next();
                byte[] value = iterator.next();
                injectionBuffer.put(ByteBufferSpanContext.ENTRY); // 1 byte
                injectionBuffer.putInt(key.length); // 4 bytes
                injectionBuffer.putInt(value.length); // 4 bytes
                injectionBuffer.put(key); // key.length
                injectionBuffer.put(value); // value.length

            }
            injectionBuffer.put(ByteBufferSpanContext.NO_ENTRY);
        } else {
            throw new IllegalArgumentException("Unsupported format: " + format);
        }
    }

    @Override
    public <C> SpanContext extract(Format<C> format, C carrier) {
        SpanContext spanContext = extractContext(format, carrier);
        if ("".equals(spanContext.toTraceId())) {
            return null;
        }
        return spanContext;
    }

    <C> SpanContext extractContext(Format<C> format, C carrier) {
        SpanContext spanContext;

        if (format.equals(Format.Builtin.TEXT_MAP)
                || format.equals(Format.Builtin.TEXT_MAP_EXTRACT)
                || format.equals(Format.Builtin.HTTP_HEADERS)) {

            if (!(carrier instanceof TextMapExtract)) {
                throw new IllegalArgumentException("Unsupported payload: " + carrier);
            }
            spanContext = new TextMapContext((TextMapExtract) carrier);

            Map<String, String> baggage = new ConcurrentHashMap<String, String>();
            SimpleSpanContext simpleSpanContext = new SimpleSpanContext();
            simpleSpanContext.setTraceId(spanContext.toTraceId());
            simpleSpanContext.setSpanId(spanContext.toSpanId());
            simpleSpanContext.setBaggage(baggage);

            Iterator<Map.Entry<String, String>> iterator = spanContext.baggageItems().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, String> entry = iterator.next();
                baggage.put(entry.getKey(), entry.getValue());
            }

            return simpleSpanContext;
        } else if (format.equals(Format.Builtin.BINARY)
                || format.equals(Format.Builtin.BINARY_EXTRACT)) {

            if (!(carrier instanceof BinaryExtract)) {
                throw new IllegalArgumentException("Unsupported payload: " + carrier);
            }
            spanContext = new ByteBufferSpanContext((BinaryExtract) carrier);

        } else {
            throw new IllegalArgumentException("Unsupported format: " + format);
        }
        return spanContext;
    }

    @Override
    public void close() {
    }


}
